package com.hxk.analysis

import org.apache.spark.sql.{Row, SparkSession}
import com.hxk.users._
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}

/**
  * 要解析的文件users.txt字段含义
  * userID,用户唯一ID
  * gender, 性别
  * age, 年龄
  * occupation, 职位
  * zipcode, 邮政编码
  */
object UsersAnalysis {

  /**
    * 练习：
    * 1.生成user类型的RDD
    * 2.使用两种方式将第一步生成的RDD转成DataFrame，然后将DataFrame转成Dataset
    * 3.将第一步的RDD转成Dataset
    * @param args
    */
  def main(args: Array[String]): Unit = {

    //0.Spark SQL程序入口
    val spark = SparkSession
      .builder()
      .appName("UsersAnalysis")
      .master("local")
      .getOrCreate()

    //1.生成user类型的RDD
    val rdd= spark.sparkContext.textFile("data/users.txt").map(line => {
      val splitData = line.split("::")
      Users(splitData(0).trim.toInt, splitData(1).trim , splitData(2).trim.toInt, splitData(3).trim, splitData(4).trim.toInt)
    })
    //rdd.collect().foreach(println)  疑问1：为什么使用这个代码就报错
    //疑问2：trim是做什么的？需要加吗？

    //2.使用两种方式将第一步生成的RDD转成DataFrame，然后将DataFrame转成Dataset
    //方式1：从RDD[A <: Product]中创建
    val dataFrameOne = spark.createDataFrame(rdd)

    //方式2：RDD转DataFrame
    import spark.implicits._
    val dataFrameTwo = rdd.toDF()

    //解析中的方法
    //val schemaString = "userID gender age occupation zipcode"
    //val schema = StructType(schemaString.split(" ").map(fieldName => StructField(fieldName,StringType,true)))
    val schema =
    StructType(
        StructField("userID", IntegerType, true) ::
        StructField("gender", StringType, true) ::
        StructField("age",IntegerType , true) ::
        StructField("occupation", StringType, true) ::
        StructField("zipcode",IntegerType, true )
        :: Nil)
    val usersRDD2 = spark.sparkContext.textFile("data/users.txt").map(line => {
      val p = line.split("::")
      Row(p(0),p(1),p(2),p(3),p(4))
    })
    val dataFramethree = spark.createDataFrame(usersRDD2,schema)

    //将DataFrame转成Dataset
    val userDatasetFromDf = dataFrameOne.as[Users]

    //3.将第一步的RDD转成Dataset
    val Ds = rdd.toDS()

    spark.stop()
  }
}
